package com.doitedu.core

import com.doitedu.utils.SparkUtil
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD

/**
 * @Date: 22.7.2 
 * @Author: HANGGE
 * @qq: 598196583
 * @Tips: 学大数据 ,到多易教育
 * @Description:
 */
object C19_Join {
  def main(args: Array[String]): Unit = {
    val sc = SparkUtil.getSparkContext("join")
    val ls1 = List[String]("刘宏", "刘勇", "李磊", "任卓", "苏敏", "晨阳")
    val ls2 = List[Double](88, 99, 99, 88, 99, 99)

    /*
        val rdd1 = sc.parallelize(ls1)
        val rdd2 = sc.parallelize(ls2)
    */

    /*
        val rdd11 = rdd1.map(name => (name, name.toUpperCase()))
        val rdd22 = rdd2.map(score => (score, score * 10))
    */

    /**
     * join 算子
     * KV数据
     *   1 根据k进行关联 , 只要K相同的元素能关联上
     *   2 有可能会出现shuffle
     *      1) 什么时候shuffle  什么不shuffle
     *
     *   3 结果RDD的分区个数是大的RDD的分区数
     */

    val ls3 = List[(String, Int)](("doit02", 2), ("doit03", 3), ("doit04", 4) ,("doit01", 1))
    val ls4 = List[(String, String)](("doit01", "李磊"), ("doit02", "刘勇"), ("doit02", "刘宏"), ("doit03", "须明"), ("doit05", "晨阳"))

    val rdd1 = sc.parallelize(ls3 , 3) //  所有可用核数资源
    val rdd2 = sc.parallelize(ls4 , 2)  // 所有可用核数资源

    println(rdd1.getNumPartitions)
    println(rdd2.getNumPartitions)

    val resRDD: RDD[(String, (Int, String))] = rdd1.join(rdd2)
    println("结果RDD的分区个数: "+ resRDD.getNumPartitions)
    resRDD.foreach(println)


    sc.stop
  }


  private def testZip(sc: SparkContext, ls1: List[String], ls2: List[Double]): RDD[(String, Double)] = {
    /**
     *
     * zip
     * 注意;
     * 两个zip的RDD
     * 1 分区个数一致
     * 2 分区内的元素个数一致
     * (两个rdd处理的数据的分区数据和分区数一致)
     */
    val rdd1 = sc.parallelize(ls1)
    val rdd2 = sc.parallelize(ls2)
    val resRDD = rdd1.zip(rdd2)
    resRDD
  }
}
